package spring.cloud.tasks.tasks_manager.waitingPreAssignTasks;

import com.alibaba.fastjson2.JSON;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.tasks_manager.schedule.TaskInstanceInfo;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
public class SlidingTimeTaskBucketQueueService {
    private final SlidingTimeTaskBucketQueue slidingTimeTaskBucketQueue = new SlidingTimeTaskBucketQueue();
    /**
     * 每次并发调度的应用数量
     */
    private static final int MAX_APP_NUM = 8;

    public void scheduleTask(List<TaskInstanceInfo> taskInstanceInfoList) {
        long nowTime = System.currentTimeMillis();
        Lists.partition(taskInstanceInfoList, MAX_APP_NUM).parallelStream().forEach(taskItemList1 -> {
            try {
                //推入时间轮中等待调度执行
                taskItemList1.forEach(taskInfo -> {
                    long nextFireDate = taskInfo.getNextFireDate().getTime();
                    long delay = 0;
                    if (nextFireDate < nowTime) {
                        log.warn("超期未执行的任务:" + JSON.toJSONString(taskInfo));
                    } else {
                        delay = nextFireDate - nowTime;
                    }
                    scheduleTask(taskInfo, delay);
                });
            } catch (Exception e) {
                log.error("schedule failed." + e.getMessage(), e);
            }
        });
    }

    //如果立即执行，那么需要真正执行的时候必须大于理论上的触发时间
    private static final long LONG_DELAY_THRESHOLD_MS = 5 * 1000;

    public void scheduleTask(TaskInstanceInfo taskInstanceInfo, Long delayMS) {
        if (delayMS <= LONG_DELAY_THRESHOLD_MS) {
            realSchedule(taskInstanceInfo, delayMS);
            return;
        } else {
            List<TaskInstanceInfo> taskInfoList = new ArrayList<>();
            taskInfoList.addAll(taskInfoList);
            slidingTimeTaskBucketQueue.addTask(taskInfoList);
        }
    }

    private void realSchedule(final TaskInstanceInfo taskInfo, Long delayMS) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                long nowTime = System.currentTimeMillis();
                //
                long delayMS = taskInfo.getNextFireDate().getTime() + 1/**保证能够能够实现业务对时间上的准确性*/ - nowTime;
                if (delayMS > 0) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(delayMS);//进入休眠状态 时间到了后不会立即进入可执行状态。
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println(taskInfo);
            }
        }).start();
    }

    /**
     * 任务取消的可能性比较低
     */
    private void cancelTask(List<TaskInstanceInfo> taskInfoList) {
        for (TaskInstanceInfo taskInfo : taskInfoList) {
            taskInfo.removeFromSlidingTimeTaskBucket();
        }
    }

    public List<TaskInstanceInfo> getTaskInstanceInfoList() {
        List<TaskInstanceInfo> list = new ArrayList<>();
        for (TaskInstanceInfo taskInstanceInfo : slidingTimeTaskBucketQueue.getTaskInstanceInfoList()) {
            if (!taskInstanceInfo.isCancelled()) {
                list.add(taskInstanceInfo);
            }
        }
        return list;
    }

}
